use super::{
    event_list_new, timer_rbtree_new, BoxedPollImpl, Event, Poll, PollEvent, PollImpl, Scheduler,
    Timer, Waker, POLLIN, POLLOUT,
};
use crate::{time, CacheLineAligned, Result};
use core::marker::PhantomData;
use core::ptr::{self, NonNull};
use core::time::Duration;
use hicollections::{List, RbTree};
use hioff::{container_of, container_of_mut};
use hipool::{Allocator, Boxed};

struct StopEvent {
    waker: Waker,
    event: Event,
    stopped: bool,
}

struct Sched<'a, A: Allocator> {
    mark: PhantomData<&'a A>,
}

#[repr(C)]
pub struct SchedImpl<'a, A: Allocator> {
    sched: Sched<'a, A>,
    poll: BoxedPollImpl<'a, A>,
    events: List<Event>,
    hi_pos: Option<NonNull<Event>>,
    timers: RbTree<Timer>,
    private_data: *const (),
    now: Duration,
    stop: StopEvent,
}

impl<'a, A: Allocator + 'a> SchedImpl<'a, A> {
    pub fn new_in(pool: &'a A) -> Result<Boxed<'a, Self, A>> {
        let waker = Waker::new()?;
        let mut boxed = Boxed::new_in(
            pool,
            CacheLineAligned::new(Self {
                sched: Sched { mark: PhantomData },
                stop: StopEvent {
                    event: Event::new(Self::stop_event_handle),
                    waker,
                    stopped: false,
                },
                poll: PollImpl::new_in(pool)?,
                events: event_list_new(),
                hi_pos: None,
                timers: timer_rbtree_new(),
                now: time::now(),
                private_data: ptr::null(),
            }),
        )?;
        boxed.active_stop_event();
        // 注意内存布局，base必须在最前面，且是#[repr(C)]，避免编译器修改
        Ok(unsafe { boxed.cast_unchecked::<Self>() })
    }
}

impl<'a, A: Allocator + 'a> SchedImpl<'a, A> {
    fn active_stop_event(&mut self) {
        let _ = self.poll.add_event(
            self.stop.waker.fd.fd(),
            POLLIN,
            ptr::addr_of!(self.stop.event) as u64,
        );
    }

    fn stop_event_handle(event: &Event, _events: u32, _flags: u16, _sched: &mut dyn Scheduler) {
        let this = unsafe { container_of_mut!(event, Self, stop.event) };
        if this.stop.waker.awaken() {
            this.stop.stopped = true;
        }
    }

    fn wait_timeout(&self) -> i32 {
        if !self.events.empty() {
            0
        } else if let Some(first) = self.timers.first() {
            let timeout = first.timeout.get();
            if timeout <= self.now {
                0
            } else {
                (timeout - self.now).as_millis() as i32
            }
        } else {
            -1
        }
    }

    fn dispatch_events(&mut self) {
        let mut list = event_list_new();
        self.events.move_head(&mut list);
        self.hi_pos = None;
        while let Some(first) = list.first() {
            unsafe { list.del(first) };
            first.handle(0, 0, self);
        }
    }

    fn dispatch_timers(timers: &mut RbTree<Timer>, sched: &mut dyn Scheduler, now: Duration) {
        while let Some(first) = timers.first() {
            if first.timeout.get() <= now {
                timers.remove(first);
                first.handle(sched)
            } else {
                break;
            }
        }
    }
}

impl<'a, A: Allocator + 'a> SchedImpl<'a, A> {
    pub fn run(&mut self) {
        while !self.stop.stopped {
            self.now = time::now();
            let timeout = self.wait_timeout();
            let iter = self.poll.wait(timeout);
            if timeout > 0 {
                self.now = time::now();
            }
            Self::dispatch_timers(&mut self.timers, &mut self.sched, self.now);

            // 对于上层异步框架而言，handle只是唤醒某个Task进入调度队列，不存在在handle的时候删除某个句柄的行为.
            for PollEvent(events, data) in iter {
                let mut events = events & (POLLIN | POLLOUT);
                if events == 0 {
                    events = POLLIN | POLLOUT;
                }
                let event = unsafe { &mut *((data & !0xFFFF_0000_0000_0000) as *mut Event) };
                let flags = (data >> 48) as u16;
                event.handle(events, flags, &mut self.sched);
            }

            self.dispatch_events();
        }
    }
}

impl<'a, A: Allocator> Scheduler for SchedImpl<'a, A> {
    fn now(&self) -> Duration {
        self.now
    }

    fn stop(&self) {
        self.stop.waker.wake();
    }

    fn private_data(&self) -> *const () {
        self.private_data
    }

    fn set_private_data(&mut self, data: *const ()) {
        self.private_data = data;
    }
    /// # Safety
    /// 调用之后e不能转移所有权
    unsafe fn add_event(&mut self, e: &Event, priority: i32) {
        self.del_event(e);
        if priority == 0 {
            self.events.add_tail(e);
        } else if let Some(hi_pos) = self.hi_pos {
            self.events.add_after(e, hi_pos.as_ref());
            self.hi_pos = Some(e.into());
        } else {
            self.events.add_head(e);
            self.hi_pos = Some(e.into());
        }
    }

    /// # Safety
    /// e应该调用了add_event
    unsafe fn del_event(&mut self, e: &Event) {
        if self.hi_pos == Some(e.into()) {
            // #14. rev_iter_from(e).next返回的仍然是e本身，应该调用nth(1)取得其前一个
            self.hi_pos = self.events.rev_iter_from(e).nth(1).map(|e| e.into());
        }
        self.events.del(e);
    }

    /// # Safety
    /// 调用之后e不能转移所有权
    unsafe fn add_fd_event(&mut self, e: &Event, events: u32, fd: i32, flags: u16) -> Result<()> {
        self.poll.add_event(fd, events, e as *const Event as u64 | ((flags as u64) << 48))
    }

    /// # Safety
    /// e应该调用了add_fd_event
    unsafe fn mod_fd_event(&mut self, e: &Event, events: u32, fd: i32, flags: u16) -> Result<()> {
        self.poll.mod_event(fd, events, e as *const Event as u64 | ((flags as u64) << 48))
    }

    /// # Safety
    /// e应该调用了add_fd_event
    unsafe fn del_fd_event(&mut self, fd: i32) -> Result<()> {
        //self.fdset.del_fd_event(e, fd, self.poll.as_mut())
        self.poll.del_event(fd)
    }

    /// # Safety
    /// 调用之后t不能转移所有权
    unsafe fn set_timer(&mut self, t: &Timer, micros: u32) {
        self.timers.remove(t);
        t.timeout
            .set(self.now + Duration::from_micros(micros as u64));
        self.timers.insert(t, false);
    }

    /// # Safety
    /// t应该调用set_timer了
    unsafe fn del_timer(&mut self, t: &Timer) {
        self.timers.remove(t);
    }
}

impl<'a, A: Allocator> Scheduler for Sched<'a, A> {
    fn now(&self) -> Duration {
        let this = unsafe { container_of!(self, SchedImpl<'a, A>, sched) };
        this.now()
    }

    fn stop(&self) {
        let this = unsafe { container_of!(self, SchedImpl<'a, A>, sched) };
        this.stop()
    }

    fn private_data(&self) -> *const () {
        let this = unsafe { container_of!(self, SchedImpl<'a, A>, sched) };
        this.private_data()
    }

    fn set_private_data(&mut self, private: *const ()) {
        let this = unsafe { container_of_mut!(self, SchedImpl<'a, A>, sched) };
        this.set_private_data(private)
    }

    unsafe fn add_event(&mut self, e: &Event, priority: i32) {
        let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
        this.add_event(e, priority)
    }

    unsafe fn del_event(&mut self, e: &Event) {
        let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
        this.del_event(e)
    }

    unsafe fn add_fd_event(&mut self, e: &Event, events: u32, fd: i32, flags: u16) -> Result<()> {
        let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
        this.add_fd_event(e, events, fd, flags)
    }

    unsafe fn mod_fd_event(&mut self, e: &Event, events: u32, fd: i32, flags: u16) -> Result<()> {
        let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
        this.mod_fd_event(e, events, fd, flags)
    }

    unsafe fn del_fd_event(&mut self, fd: i32) -> Result<()> {
        let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
        this.del_fd_event(fd)
    }

    unsafe fn set_timer(&mut self, t: &Timer, msecs: u32) {
        let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
        this.set_timer(t, msecs)
    }

    unsafe fn del_timer(&mut self, t: &Timer) {
        let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
        this.del_timer(t)
    }
}
